package com.simon.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class CustomProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

    private String  name;

    public CustomProcessor() {
        this.name = "default processor";
    }

    public CustomProcessor(String name) {
        this.name = name;
    }

    private Flow.Subscription subscription;

    // @Override
    // public void subscribe(Flow.Subscriber<? super String> subscriber) {
    //
    // }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println(this.name + ".onSubscribe() >>> start...");
        this.subscription = subscription;

        this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println(this.name + ".onNext() >>> item: " + item);
        item += " processor";
        submit(item);
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println(this.name + ".onError() >>> start...");
    }

    @Override
    public void onComplete() {
        System.out.println(this.name + ".onComplete() >>> start...");
    }
}
